package cn.baiyang.rocketmq.spring.core;

import java.nio.charset.StandardCharsets;
import java.util.Map;
import java.util.Objects;

import org.apache.rocketmq.client.apis.ClientException;
import org.apache.rocketmq.client.apis.producer.Producer;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.core.env.Environment;

import io.github.bucket4j.distributed.proxy.ProxyManager;

/**
 *
 * @author hongzhu
 * @version V1.0
 * @since 2024-03-13 10:00
 */
public class RocketMQProducerTemplate extends RocketMQSender implements InitializingBean, DisposableBean {

    @Override
    public void setEnvironment(Environment environment) {
        this.environment = environment;
    }

    @Override
    public void setProxyManager(ProxyManager<String> proxyManager) {
        this.proxyManager = proxyManager;
    }

    @Override
    public void setProducer(Producer producer) {
        this.producer = producer;
    }

    @Override
    public void destroy() throws Exception {
        if (Objects.nonNull(producer)) {
            producer.close();
        }
    }

    @Override
    public void afterPropertiesSet() throws Exception {

    }

    public String send(String topic, String body) throws ClientException {
        return send(topic, null, body);
    }

    public String send(String topic, String tag, String body) throws ClientException {
        return send(topic, tag, body, null);
    }

    public String send(String topic, String tag, String body, RocketMQMessageParameters parameters) throws ClientException {
        return send(topic, tag, body.getBytes(StandardCharsets.UTF_8), parameters);
    }

    public String send(String topic, byte[] body) throws ClientException {
        return send(topic, null, body);
    }

    public String send(String topic, String tag, byte[] body) throws ClientException {
        return send(topic, tag, body, null);
    }

    public String sendAsync(String topic, String body) throws ClientException {
        return sendAsync(topic, null, body);
    }

    public String sendAsync(String topic, String tag, String body) throws ClientException {
        return sendAsync(topic, tag, body, null);
    }

    public String sendAsync(String topic, String tag, String body, RocketMQMessageParameters parameters) throws ClientException {
        return sendAsync(topic, tag, body.getBytes(StandardCharsets.UTF_8), parameters);
    }

    public String sendAsync(String topic, byte[] body) throws ClientException {
        return sendAsync(topic, null, body);
    }

    public String sendAsync(String topic, String tag, byte[] body) throws ClientException {
        return sendAsync(topic, tag, body, null);
    }

    public String send(String topic, String tag, byte[] body, RocketMQMessageParameters parameters) throws ClientException {
        String messageGroup = null == parameters ? null : parameters.getMessageGroup();
        Long deliveryTimestamp = null == parameters ? null : parameters.getDeliveryTimestamp();
        Map<String, String> properties = null == parameters ? null : parameters.getProperties();
        String[] keys = null == parameters ? null : parameters.getKeys();
        return send(buildMessage(topic, tag, messageGroup, body, deliveryTimestamp, properties, keys));
    }

    public String sendAsync(String topic, String tag, byte[] body, RocketMQMessageParameters parameters) {
        String messageGroup = null == parameters ? null : parameters.getMessageGroup();
        Long deliveryTimestamp = null == parameters ? null : parameters.getDeliveryTimestamp();
        Map<String, String> properties = null == parameters ? null : parameters.getProperties();
        String[] keys = null == parameters ? null : parameters.getKeys();
        return sendAsync(buildMessage(topic, tag, messageGroup, body, deliveryTimestamp, properties, keys));
    }

}
